package org.alanlau.consumer;

import com.rabbitmq.client.Channel;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * 消息签收
 *
 * @author alan
 * @version 1.0
 * @date 2021/4/8 18:12
 */
@Component
public class MsgAckCustomer {
    /**
     * 用来记录消费失败的消息，重复消息的次数，超过一定次数，直接删除消息，记录日志即可，防止死循环
     * 注意：实际生产环境中，一般都是分布式环境，所以肯定是缓存到redis里来做这个操作
     */
    Map<String, Integer> cache = new ConcurrentHashMap<>();

    @RabbitListener(queues = "boot.msg.act.queue")
    public void receiveMessage(String content, Message message, Channel channel) throws IOException {
        long deliveryTag = message.getMessageProperties().getDeliveryTag();
        String messageId = message.getMessageProperties().getMessageId();

        System.out.println("消息投递ID → ：" + deliveryTag);
        System.out.println("消息自定义ID → ：" + messageId);

        if (content.contains("正确")) {
            channel.basicAck(deliveryTag, true);
            System.out.println("消息签收成功 → 内容为：" + content);
        } else {
            Integer count = cache.getOrDefault(messageId, 1);
            if (count >= 3) {
                channel.basicNack(deliveryTag, false, false);
                System.out.println("该消息消费【3】次都失败，我们记录它，人工处理" + content);
                // 既然都删除了，就把缓存的记录也清除，防止空间浪费
                cache.remove(messageId);
            } else {
                // 不签收，并扔回消息队列
                channel.basicNack(deliveryTag, false, true);
                // 计数+1
                cache.put(messageId, count + 1);
            }
        }
    }
}
